package org.iscas.tcse.util;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;

public class BlackHoleSink<T> extends RichSinkFunction<T> {
    private static final Logger LOG = LoggerFactory.getLogger(BlackHoleSink.class);
    private int count = 0;
    private String redisHost;
    private int redisPort;
    private Jedis jedis;

    private String timeKey;

    public BlackHoleSink(String redisHost, int redisPort, String name) {
        this.redisHost = redisHost;
        this.redisPort = redisPort;
        this.timeKey = name + "_blackhole_sink_time";
        Jedis client = new Jedis(redisHost, redisPort);
        client.del(timeKey + "_start");
        client.del(timeKey + "_end");
        client.close();
    }

    @Override
    public void invoke(T value, Context context) throws Exception {
        if (count == 0) {
            jedis.setnx(timeKey + "_start", String.valueOf(System.currentTimeMillis() / 10));
        }
        count++;
        if (count % 100000 == 0) {
            LOG.info("sunk count: {}", count);
            jedis.set(timeKey + "_end", String.valueOf(System.currentTimeMillis() / 10));
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        jedis = new Jedis(redisHost, redisPort);
    }

    @Override
    public void close() throws Exception {
        super.close();
        jedis.set(timeKey + "_end", String.valueOf(System.currentTimeMillis() / 10));
        jedis.close();
    }
}
